package com.hank.session;

import com.alibaba.fastjson.JSONObject;
import com.hank.Dao.TaskDao;
import com.hank.conf.ConfigurationManager;
import com.hank.constant.Constants;
import com.hank.domain.Task;
import com.hank.factory.DaoFactory;
import com.hank.mockdata.MockData;
import com.hank.util.DateUtils;
import com.hank.util.StringUtils;
import com.hank.util.ValidUtils;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.storage.StorageLevel;
import scala.Tuple2;

import java.util.Date;
import java.util.List;

/**
 * 用户行为分析主入口
 */
public class UserVisitAnalyze {
    public static void main(String[] args) {
            args = new String[]{"1"};

            SparkConf conf = new SparkConf().setAppName(Constants.APP_NAME_SESSION).setMaster("local");

            JavaSparkContext context = new JavaSparkContext(conf);
            SQLContext sc = getSQLContext(context.sc());

            //载入模拟数据
            mock(context, sc);

            //拿到相应的Dao组件
            TaskDao taskDao = DaoFactory.getTaskDao();
            //从外部传入得参数获取任务id
//        Long taskId = Long.valueOf(args[0]);
            Long taskId = Long.valueOf("1");

            //从数据库中查询出相应的task
            Task task = taskDao.findTaskById(taskId);
            JSONObject jsonObject = JSONObject.parseObject(task.getTaskParam());

            //获取指定范围内的Session
            JavaRDD<Row> sessionRangeDate = getActionRDD(sc, jsonObject);

            //转换成键值对
            JavaPairRDD<String, Row> sessionInfoPairRDD = getSessionInfoPairRDD(sessionRangeDate);

            //将重复用的RDD进行持久化 TODO HANK 这里感觉不需要持久化
            sessionInfoPairRDD.persist(StorageLevel.MEMORY_AND_DISK());

            //按照Session进行聚合
            JavaPairRDD<String, String> sesssionAggregateInfoRDD = aggregateBySessionId(sc, sessionInfoPairRDD);

        //通过条件对RDD进行筛选
        Accumulator<String> sessionAggrStatAccumulator=context.accumulator("",new SessionAggrStatAccumulator());

        //在进行accumulator之前，需要action动作,不然为空(设置计数器)
        JavaPairRDD<String,String> filteredSessionRDD=filterSessionAndAggrStat(sesssionAggregateInfoRDD,jsonObject,sessionAggrStatAccumulator);

        filteredSessionRDD.persist(StorageLevel.MEMORY_AND_DISK());

        //获取符合过滤条件的全信息公共RDD
        JavaSparkContext<>

        context.close();

    }

    /**
     * 用于判断是否生产环境
     */
    public static SQLContext getSQLContext(SparkContext sc)
    {
        boolean local= ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
        if(local){
            return new SQLContext(sc);
        }
        return new HiveContext(sc);
    }

    /**
     * 模拟数据生成
     * @param context
     * @param sc
     */
    private static void mock(JavaSparkContext context,SQLContext sc)
    {
        boolean local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
        if(local){
            MockData.mock(context,sc);
        }
    }

    /**
     * 获取指定日期范围内的数据
     * @param sc
     * @param taskParam
     * @return
     */
    private static JavaRDD<Row> getActionRDD(SQLContext sc,JSONObject taskParam){

//        String startTime = ParamUtils.getParam(taskParam,Constants.PARAM_STARTTIME);
//        String endTime  = ParamUtils.getParam(taskParam,Constants.PARAM_ENDTIME);

        String startTime = "2019-06-10";
        String endTime = "2019-06-14";
        String sql="select * from user_visit_action where date>='"+startTime+"' and date<='"+endTime+"'";
        DataFrame df= sc.sql(sql);
//        for(Row row:df.take(30)){
//            System.out.println(row);
//        }
        return df.javaRDD();
    }

    /**
     * 将数据进行映射成为Pair,键为SessionId,Value为Row
     * @param sessionRangeDate
     * @return
     */
    private static JavaPairRDD<String,Row> getSessionInfoPairRDD(JavaRDD<Row> sessionRangeDate){
        return sessionRangeDate.mapToPair(new PairFunction<Row, String, Row>() {
            @Override
            public Tuple2<String, Row> call(Row row) throws Exception {
                return new Tuple2<>(row.getString(2),row);
            }
        });
    }

    /**
     * session粒度的聚合
     * @param sc
     * @param sessionInfoPairRDD
     * @return
     */
    private static JavaPairRDD<String,String> aggregateBySessionId(SQLContext sc,JavaPairRDD<String,Row> sessionInfoPairRDD){
        //根据sessionId进行分组
       JavaPairRDD<String,Iterable<Row>> sessionActionGrouped= sessionInfoPairRDD.groupByKey();

       JavaPairRDD<Long,String> sessionPartInfo = sessionActionGrouped.mapToPair(new PairFunction<Tuple2<String, Iterable<Row>>, Long, String>() {
           @Override
           public Tuple2<Long, String> call(Tuple2<String, Iterable<Row>> stringIterableTuple2) throws Exception {
               String sessionId = stringIterableTuple2._1;
               Iterable<Row> rows = stringIterableTuple2._2;

               StringBuffer searchKeywords = new StringBuffer();
               StringBuffer clickCategoryIds = new StringBuffer();
               Long userId=null;
               Date startTime=null;
               Date endTime=null;
               int stepLength=0;

               //处理数据集合
               for(Row row: rows){
                    if (userId == null){
                        userId=row.getLong(1);
                   }
                   String searchKeyword=row.getString(5);
                    Long clickCategoryId =row.getLong(6);

                    if(StringUtils.isNotEmpty(searchKeyword)){
                        if(!searchKeywords.toString().contains(searchKeyword)){
                            searchKeywords.append(searchKeyword+",");
                        }
                    }

                    if(clickCategoryId != null){
                        if(!clickCategoryIds.toString().contains(String.valueOf(clickCategoryId))){
                            clickCategoryIds.append(String.valueOf(clickCategoryId)+",");
                        }
                    }

                    //计算session开始时间和结束时间
                   Date actionTime = DateUtils.parseTime(row.getString(4));
                    if(startTime == null){
                        startTime= actionTime;
                    }
                    if(endTime== null){
                        endTime=actionTime;
                    }
                    if(actionTime.before(startTime)){
                        startTime=actionTime;
                    }
                    if(actionTime.after(endTime)){
                        endTime=actionTime;
                    }

                    //步骤长度
                   stepLength++;
               }

               //访问时长(s)
               Long visitLength=(endTime.getTime()-startTime.getTime())/1000;

               String searchKeywordsInfo=StringUtils.trimComma(searchKeywords.toString());
               String clickCategoryIdsInfo=StringUtils.trimComma(clickCategoryIds.toString());

               String info=Constants.FIELD_SESSIONID+"="+sessionId+"|"+Constants.FIELD_SERACH_KEYWORDS+"="+searchKeywordsInfo+"|"
                       +Constants.FIELD_CLICK_CATEGORYIDS+"="+clickCategoryIdsInfo+"|"+Constants.FIELD_VISIT_LENGTH+"="+visitLength+"|"
                       +Constants.FIELD_STEP_LENGTH+"="+stepLength+"|"+Constants.FIELD_START_TIME+"="+DateUtils.formatTime(startTime);
                return new Tuple2<Long,String>(userId,info);
           }
       });

        //查询所有的用户数据
        String sql="select * from user_info";
        JavaRDD<Row> userInfoRDD=sc.sql(sql).javaRDD();
        //将用户信息映射成map
        JavaPairRDD<Long,Row> userInfoPariRDD=userInfoRDD.mapToPair(new PairFunction<Row, Long, Row>() {
            @Override
            public Tuple2<Long, Row> call(Row row) throws Exception {
                return new Tuple2<>(row.getLong(0),row);
            }
        });

        //将两个信息join在一起 TODO优化 HANK
        JavaPairRDD<Long,Tuple2<String,Row>> tuple2JavaPairRDD=sessionPartInfo.join(userInfoPariRDD);

        /**
         * 拿到所需要的session
         */
        JavaPairRDD<String,String> sessionInfo=tuple2JavaPairRDD.mapToPair(new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> longTuple2Tuple2) throws Exception {
                String sessionPartInfo = longTuple2Tuple2._2._1;
                Row userInfo =longTuple2Tuple2._2._2;
                //拿到需要的用户信息
                int age =userInfo.getInt(3);
                String professional=userInfo.getString(4);
                String city=userInfo.getString(5);
                String sex=userInfo.getString(6);
                //拼接字符串
                String fullInfo=sessionPartInfo+"|"+Constants.FIELD_AGE+"="+age+"|"
                        +Constants.FIELD_PROFESSIONAL+"="+professional+"|"+Constants.FIELD_CITY+"="+city+"|"+Constants.FIELD_SEX+"="+sex;
                String session=StringUtils.getFieldFromConcatString(sessionPartInfo,"\\|",Constants.FIELD_SESSIONID);
                return new Tuple2<String, String>(session,fullInfo);
            }
        });
        return sessionInfo;
    }

    private static JavaPairRDD<String,String>filterSessionAndAggrStat(JavaPairRDD<String, String> sessionInfoRDD,final JSONObject taskParam,final Accumulator<String> sessionAggrStatAccumulator){
        //得到条件
//        String startAge =ParamUtils.getParam(taskParam,Constants.PARAM_STARTAGE);
//        String endAge =ParamUtils.getParam(taskParam,Constants.PARAM_ENDAGE);
//        String professionals=ParamUtils.getParam(taskParam,Constants.PARAM_PROFESSONALS);
//        String cities=ParamUtils.getParam(taskParam,Constants.PARAM_CIYTIES);
//        String sex = ParamUtils.getParam(taskParam,Constants.PARAM_SEX);
//        String keyWords=ParamUtils.getParam(taskParam,Constants.PARAM_SERACH_KEYWORDS);
//        String categoryIds=ParamUtils.getParam(taskParam,Constants.PARAM_CLICK_CATEGORYIDS);

        String startAge =null;
        String endAge =null;
        String professionals=null;
        String cities=null;
        String sex =null;
        String keyWords=null;
        String categoryIds=null;

        //拼接参数
        String _paramter=(startAge!=null?Constants.PARAM_STARTAGE+"="+startAge+"|":"")+
                (endAge!=null?Constants.PARAM_ENDAGE+"="+endAge+"|":"")+(professionals!=null?Constants.PARAM_PROFESSONALS+"="+professionals+"|":"")+
                (cities!=null?Constants.PARAM_CIYTIES+"="+cities+"|":"")+(sex!=null?Constants.PARAM_SEX+"="+sex+"|":"")+
                (keyWords!=null?Constants.PARAM_SERACH_KEYWORDS+"="+keyWords+"|":"")+(categoryIds!=null?Constants.PARAM_CLICK_CATEGORYIDS+"="+categoryIds+"|":"");


        if(_paramter.endsWith("\\|")){
            _paramter=_paramter.substring(0,_paramter.length()-1);
        }

        final String paramter=_paramter;

        JavaPairRDD<String,String> filterSessionRDD = sessionInfoRDD=sessionInfoRDD.filter(new Function<Tuple2<String, String>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, String> tuple2) throws Exception {
                String sessionInfo = tuple2._2;
                //按照条件进行过滤
                //按照年龄进行过滤
                if(!ValidUtils.between(sessionInfo,Constants.FIELD_AGE,paramter,Constants.PARAM_STARTAGE,Constants.PARAM_ENDAGE))
                    return  false;
                //按照职业进行过滤
                if(!ValidUtils.in(sessionInfo,Constants.FIELD_PROFESSIONAL,paramter,Constants.PARAM_PROFESSONALS))
                    return false;
                //按照城市进行过滤
                if(!ValidUtils.in(sessionInfo,Constants.FIELD_CITY,paramter,Constants.PARAM_CIYTIES))
                    return false;
                //按照性别进行筛选
                if(!ValidUtils.equal(sessionInfo,Constants.FIELD_SEX,paramter,Constants.PARAM_SEX))
                    return false;
                //按照搜索词进行过滤，只要有一个搜索词即可
                if(!ValidUtils.in(sessionInfo,Constants.FIELD_SERACH_KEYWORDS,paramter,Constants.PARAM_PROFESSONALS))
                    return false;
                if(!ValidUtils.in(sessionInfo,Constants.FIELD_CLICK_CATEGORYIDS,paramter,Constants.FIELD_CLICK_CATEGORYIDS))
                    return false;

                //如果经过了之前的所有的过滤条件，也就是满足用户筛选条件
                sessionAggrStatAccumulator.add(Constants.SESSION_COUNT);
                //计算出访问时长和访问步长的范围并进行相应的累加
                Long visitLength=Long.valueOf(StringUtils.getFieldFromConcatString(sessionInfo,"\\|",Constants.FIELD_VISIT_LENGTH));
                Long stepLength=Long.valueOf(StringUtils.getFieldFromConcatString(sessionInfo,"\\|",Constants.FIELD_STEP_LENGTH));

                //使用函数进行统计
                calculateVisitLength(visitLength);
                calculateStepLength(stepLength);

                return true;
            }

            //统计访问时长的数量
            private void calculateVisitLength(Long visitLegth)
            {
                if(visitLegth>=1&&visitLegth<=3)
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1s_3s);
                else if(visitLegth>=4&&visitLegth<=6)
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_4s_6s);
                else if(visitLegth>=7&&visitLegth<=9)
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_7s_9s);
                else if(visitLegth>=10&&visitLegth<=30)
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10s_30s);
                else if(visitLegth>30&&visitLegth<=60)
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30s_60s);
                else if(visitLegth>60&&visitLegth<=180)
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1m_3m);
                else if(visitLegth>180&&visitLegth<=600)
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_3m_10m);
                else if(visitLegth>600&&visitLegth<=1800)
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10m_30m);
                else if(visitLegth>1800)
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30m);
            }

            //统计访问步长的数量
            private void calculateStepLength(Long stepLength)
            {
                if(stepLength>=1&&stepLength<=3)
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_1_3);
                else if(stepLength>=4&&stepLength<=6)
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_4_6);
                else if(stepLength>=7&&stepLength<=9)
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_7_9);
                else if(stepLength>=10&&stepLength<=30)
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_10_30);
                else if(stepLength>30&&stepLength<=60)
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_30_60);
                else if(stepLength>60)
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_60);
            }
        });
        return filterSessionRDD;
    }

    public static JavaPairRDD<String,Row> getFilterFullInfoRDD(JavaPairRDD<String,String> )
}
